package _kafka;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
/*
* 如果是SSL接入点实例，请取消以下三行代码的注释；如果是SASL接入点，请取消以下前两行代码的注释。
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaConsumerDemo {

    public static void main(String args[]) {

        //设置JAAS配置文件的路径。
        /*
         * 如果是SSL接入点实例，请取消以下一行代码的注释。
        JavaKafkaConfigurer.configureSasl();
         */
                        
        /*
         * 如果是SASL接入点PLAIN机制实例，请取消以下一行代码的注释。
        JavaKafkaConfigurer.configureSaslPlain();
         */
                        
        /*
        * 如果是SASL接入点SCRAM机制实例，请取消以下一行代码的注释。
        JavaKafkaConfigurer.configureSaslScram();
        */

        //加载kafka.properties
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        //设置接入点，请通过控制台获取对应Topic的接入点。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

        //如果是SSL接入点实例，请注释以下第一行代码。
        //可更加实际拉去数据和客户的版本等设置此值，默认30s。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * 如果是SSL接入点实例，请取消以下六行代码的注释。
         * 设置SSL根证书的路径，请记得将XXX修改为自己的路径。
         * 与SASL路径类似，该文件也不能被打包到jar中。
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * 根证书存储的密码，保持不变。
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * 接入协议，目前支持使用SASL_SSL协议接入。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * SASL鉴权方式，保持不变。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * 两次Poll之间的最大允许间隔。
         * 消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Group移除并触发Rebalance，默认30s。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         * 设置单次拉取的量，走公网访问时，该参数会有较大影响。
         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
         */

        //如果是SASL接入点PLAIN机制实例，请注释以下一行代码。
        //可更加实际拉去数据和客户的版本等设置此值，默认30s。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * 如果是SASL接入点PLAIN机制实例，请取消以下三行代码的注释。
         * 接入协议。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * Plain方式。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * 两次Poll之间的最大允许间隔。
         * 消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Group移除并触发Rebalance，默认30s。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        //如果是SASL接入点SCRAM机制实例，请注释以下一行代码。
        //可更加实际拉去数据和客户的版本等设置此值，默认30s
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * 如果是SASL接入点SCRAM机制实例，请取消以下四行代码的注释。
         * 接入协议。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * SCRAM方式。
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         * 两次Poll之间的最大允许间隔。
         * 消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Group移除并触发Rebalance，默认30s。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        //每次poll的最大数量。
        //注意该值不要改得太大，如果poll太多数据，而不能在下次poll之前消费完，则会触发一次负载均衡，产生卡顿。
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        //消息的反序列化方式
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //当前消费实例所属的消费组，请在控制台申请之后填写。
        //属于同一个组的消费实例，会负载消费消息。
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));

        //如果是SSL接入点实例，请取消以下一行代码的注释。
        //Hostname校验改成空。
        //props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        //构造消息对象，也即生成一个消费实例。
        KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
        //设置消费组订阅的Topic，可以订阅多个。
        //如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样。
        List<String> subscribedTopics =  new ArrayList<String>();

        //如果是SSL接入点实例，请注释以下前五行代码，取消第六行代码的注释。
        //如果需要订阅多个Topic，则在这里add进去即可。
        //每个Topic需要先在控制台进行创建。
        String topicStr = kafkaProperties.getProperty("topic");
        String[] topics = topicStr.split(",");
        for (String topic: topics) {
            subscribedTopics.add(topic.trim());
        }
        //subscribedTopics.add(kafkaProperties.getProperty("topic"));
        consumer.subscribe(subscribedTopics);

        //循环消费消息。
        while (true){
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                //必须在下次poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。
                //建议开一个单独的线程池来消费消息，然后异步返回结果。
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(String.format("Consume partition:%d, offset:%d, value:%d", record.partition(), record.offset(), record.value()));
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (Throwable ignore) {

                }

                e.printStackTrace();
            }
        }
    }
}